package cn.itcast.czxy

import java.util.Properties

import cn.itcast.czxy.BD18.bean.{HBaseMeta, TagRule}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
 * 政治面貌标签
 *
 *
 */
object PoliticalFaceTag {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().appName("PoliticalFaceTag").master("local[*]").getOrCreate()


    //2链接数据库
    var url="jdbc:mysql://bd001:3306/tags_new?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"//修改数据库名称
    var table="tbl_basic_tag"
    var properties=new Properties
    val mysqlConn: DataFrame = spark.read.jdbc(url,table,properties)


    //隐式转换
    import spark.implicits._
    import scala.collection.JavaConverters._
    import org.apache.spark.sql.functions._


    //3读取四级标签
    val fourDS: Dataset[Row] = mysqlConn.select("rule").where("id=74")//修改  id

    val fourMap: Map[String, String] = fourDS.map(row => {
      //使用##  切分再使用=切分
      row.getAs("rule").toString.split("##")
        .map(kv => {
          val arr: Array[String] = kv.split("=")
          (arr(0), arr(1))
        })
    }).collectAsList().get(0).toMap

    //转换
    var hbaseMeta:HBaseMeta=getHBaseMeta(fourMap)


    //4读取五级标签
    val fiveDS: Dataset[Row] = mysqlConn.select("id","rule").where("pid=74")//修改  id
    val fiveTagRule: List[TagRule] = fiveDS.map(row => {
      //获取数据
      val id: Int = row.getAs("id").toString.toInt
      val rule: String = row.getAs("rule").toString
      //封装
      TagRule(id, rule)
    }).collectAsList().asScala.toList


    //5、读取hbase数据
    val HBaseDatas: DataFrame = spark.read.format("cn.itcast.czxy.BD18.tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, hbaseMeta.zkHosts)
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE, hbaseMeta.hbaseTable)
      .option(HBaseMeta.FAMILY, hbaseMeta.family)
      .option(HBaseMeta.SELECTFIELDS, hbaseMeta.selectFields)
      .load()


    //政治面貌变换的自定义函数
    var  getTags=udf((rule:String)=>{
      //遍历每一个rule  判断是否与数据中的相同，若相同返回对应的ID
      var tagId=0
      for(tagRule<-fiveTagRule){
        if (tagRule.rule==rule){
          tagId=tagRule.id
        }
      }
      tagId
    })

    //6、使用五级标签与hbase数据进行匹配获得标签
    val JobNewTags: DataFrame = HBaseDatas.select('id. as ("userId"),getTags('politicalFace).as("tagsId"))
    JobNewTags.show()


    var getAllTagas=udf((oldTagsId:String,newTagsId:String)=>{
      if (oldTagsId==""){
        newTagsId
      }else if (newTagsId==""){
        oldTagsId
      }else if(oldTagsId==""&& newTagsId==""){
        ""
      }else{
        //拼接历史数据和新数据（可能有重复的数据）
        val alltags   = oldTagsId+","+newTagsId
        //对重复数据区中去重
        alltags.split(",").distinct
          //使用逗号分隔，返回字符串类型。
          .mkString(",")
      }
    })


    //7、解决数据覆盖的问题
    val oldTags: DataFrame = spark.read.format("cn.itcast.czxy.BD18.tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, hbaseMeta.zkHosts)
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE,"test")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagsId")
      .load()

    //追加新计算出来的标签到历史数据
    val joinTagas: DataFrame = oldTags.join(JobNewTags,    oldTags("userId")===JobNewTags("userId"))

    val allTags: DataFrame = joinTagas.select(
      //处理第一个字段    两个表中的多个userId字段，只读取一个
      when((oldTags.col("userId").isNotNull),(oldTags.col("userId")))
        .when((JobNewTags.col("userId").isNotNull),(JobNewTags.col("userId")))
        .as("userId"),

      //处理第二个字段  将两个字段个合并一起
      getAllTagas(oldTags.col("tagsId"),JobNewTags.col("tagsId")).as("tagsId")
    )

    //8、将最终数据写入hbase
    allTags.write.format("cn.itcast.czxy.BD18.tools.HBaseDataSource")
      .option("zkHosts", hbaseMeta.zkHosts)
      .option(HBaseMeta.ZKPORT, hbaseMeta.zkPort)
      .option(HBaseMeta.HBASETABLE,"test")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "userId,tagsId")
      .save()

  }

  //将map 转换成样HBaseMeta例类
  def getHBaseMeta(fourMap: Map[String, String]): HBaseMeta = {
    HBaseMeta(fourMap.getOrElse(HBaseMeta.INTYPE,""),
      fourMap.getOrElse(HBaseMeta.ZKHOSTS,""),
      fourMap.getOrElse(HBaseMeta.ZKPORT,""),
      fourMap.getOrElse(HBaseMeta.HBASETABLE,""),
      fourMap.getOrElse(HBaseMeta.FAMILY,""),
      fourMap.getOrElse(HBaseMeta.SELECTFIELDS,""),
      fourMap.getOrElse(HBaseMeta.ROWKEY,"")
    )
  }

}
